package cn.witsky.rtp2ws.push.disruptor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;
import cn.witsky.rtp2ws.Server;
import cn.witsky.rtp2ws.constant.ThreadPoolConstant;
import cn.witsky.rtp2ws.domain.Message;
import cn.witsky.rtp2ws.exception.IntEventExceptionHandler;
import cn.witsky.rtp2ws.util.DefaultThreadFactory;

 
public class EventDisruptor{
	private RingBuffer<Message> ringBuffer = null;

	private EventProducer producer = null;

	public EventProducer getProducer() {
		return producer;
	}

	public RingBuffer<Message> getRingBuffer() {
		return ringBuffer;
	}

	public EventDisruptor(int index) {
		// 创建缓冲池
		ThreadFactory threadFactory=new DefaultThreadFactory(ThreadPoolConstant.getDisruptorThreadName(index));
		
		ExecutorService executor = Executors.newFixedThreadPool(1,threadFactory);
		// 创建工厂
		EventFactory<Message> factory = new EventFactory<Message>() {
			@Override
			public Message newInstance() {
				return new Message();
			}
		};
		
//		WaitStrategy waitStrategy=new YieldingWaitStrategy();
		WaitStrategy waitStrategy = new MyWaitStrategy();
		// 创建ringBuffer
		ringBuffer = RingBuffer.create(ProducerType.MULTI, factory,Server.RING_BUFFER_SIZE, waitStrategy);
		SequenceBarrier barriers = ringBuffer.newBarrier();
		EventConsumer consumers = new EventConsumer();
		WorkerPool<Message> workerPool = new WorkerPool<Message>(ringBuffer, barriers,new IntEventExceptionHandler(), consumers);
		ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
		workerPool.start(executor);
		producer = new EventProducer(ringBuffer);
	}

}